/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.cluster.impl.kryo;

import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;

import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;
import org.sabayframework.channel.ram.ConcurrentArrayChannel;
import org.sabayframework.cluster.Computer;
import org.sabayframework.cluster.ComputerId;
import org.sabayframework.cluster.DistributedActivity;
import org.sabayframework.cluster.impl.SingleNodeStrategy;
import org.sabayframework.log.Log;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Output;
/**
 * Class is responsible for bookkeeping of network requests.
 * */
public class RequestManager  {
	/** activities from remote clients that are processing by this computer */
	private final List<DistributedActivityContext> localActivities;
	/** queue of requests that are ready to be sent to remote severs */
	private final List<Request> pendingRequests;
	/** activities that were sent to remote servers, key is an activity id */
	private final Map<Integer, DistributedActivityContext> remoteActivities;
	private final ReentrantLock lock;
	private final ComputerId localComputerId;
    private final int maxQueueSize;
    private final int maxBufferSize;
    private final Kryo kryo;
	
	private NetworkManager networkManager;
	
	public RequestManager(final Kryo kryo, final ReentrantLock lock, final ComputerId localComputerId, final int maxQueueSize, int maxBufferSize) {
		super();
		this.kryo = kryo;
		this.pendingRequests = new LinkedList<Request>();
		this.remoteActivities = Collections.synchronizedMap(new HashMap<Integer, DistributedActivityContext>());
		this.localActivities = Collections.synchronizedList(new LinkedList<DistributedActivityContext>());
		this.lock = lock;
		this.localComputerId = localComputerId;
		this.maxQueueSize = maxQueueSize;
		this.maxBufferSize = maxBufferSize;
	}
	
	public DistributedActivityContext addLocalActivity(ComputerId clientId, NetworkPacket data){
		DistributedActivityContext r = new DistributedActivityContext(data.getActivityId());
		r.setClientId(clientId);
		r.addServerId(localComputerId);
		r.setInput(new ConcurrentArrayChannel<Object>(maxQueueSize));
		r.setOutput(new ConcurrentArrayChannel<Object>(maxQueueSize));
		r.setDistributedActivity(new DistributedActivity<Object, Object>(
				null, // will be assigned later in RequestActivity
				new SingleNodeStrategy(localComputerId), 
				data.getImportPackage(), 
				data.getRequireBundle()));
		localActivities.add(r);
		return r;
	}
	public void submitRequest(Request r)
	{
		lock.lock();
		try{
			pendingRequests.add(r);
		}finally{
			lock.unlock();
		}
	}
	public DistributedActivityContext findLocalActivity(ComputerId clientId, NetworkPacket data){
		DistributedActivityContext rs[] = localActivities.toArray(new DistributedActivityContext[0]);
		for(DistributedActivityContext r : rs){
			if(r.getClientId().equals(clientId) && r.getActivityId() == data.getActivityId()) return r;
		}
		return null;
	}
	public DistributedActivityContext findRemoteActivity(NetworkPacket data){
		return remoteActivities.get(data.getActivityId());
	}
	
	public boolean canProcessRequests(){
		final ReentrantLock lock = this.lock;
		if(lock.tryLock()){
			try{
				if(!pendingRequests.isEmpty()) return true;
				if(!localActivities.isEmpty()){
					DistributedActivityContext[] rs = localActivities.toArray(new DistributedActivityContext[0]);
					for(DistributedActivityContext r : rs){
						final Channel<?> out = r.getOutput();
						if(out.size()>0) return true;
					}
				}
				if(!remoteActivities.isEmpty()){
					DistributedActivityContext[] rs = remoteActivities.values().toArray(new DistributedActivityContext[0]);
					for(DistributedActivityContext r : rs){
						final Channel<?> out = r.getInput();
						if(out.size()>0) return true;
					}
				}
			} finally{
				lock.unlock();
			}
		}
		return false;
	}
	public void processRequests(){
		Log.debug("Processing pending/IO requests");
		processPendingRequests();
		processLocalActivities();
		processRemoteActivities();
	}
	private void processLocalActivities(){
		// iterate through serving requests
		// for each request try to read from output channels and send to client
		DistributedActivityContext[] rs = localActivities.toArray(new DistributedActivityContext[0]);
		for(DistributedActivityContext r : rs){
			try{
				processOutputChannel(r);
			} catch (Throwable t) {
				failActivity(r, t);
			}
			final Channel<?> out = r.getOutput();
			if(out.isClosed() && out.isEmpty()) localActivities.remove(r);
		}
	}
	private void processOutputChannel(DistributedActivityContext dac) throws IOException{
		final Channel<?> out = dac.getOutput();
		final int available = out.size();
		NetworkPacket packet = null;
		if(available > 0){
			final Lock<?> rl = out.getReadLock(1);
			if(rl!=null){
				Object data = null;
				try{
					data = rl.get(0);
				}finally{
					rl.unlock();
				}
				packet = createWriteData(dac, data, RequestType.OUTPUT);
				networkManager.sendData(dac.getClientId(), packet);
			}
		}
		if(out.isClosed() && out.isEmpty()){
			packet = createCloseData(dac, out.getFailureCause(),RequestType.CLOSE_OUTPUT);
			networkManager.sendData(dac.getClientId(), packet);
		}
	}
	
	private void processPendingRequests(){
		//iterate through pending requests and try to send them to server
		final ReentrantLock lock = this.lock;
		Request[] rs = null;
		lock.lock();
		try{
			rs = pendingRequests.toArray(new Request[0]);
			pendingRequests.clear();
		}finally{
			lock.unlock();
		}
		for(Request r : rs){
			try{
				Log.debug("Processing pending request: ", r);
				DistributedActivityContext dac = r.getDistributedActivityContext();
				switch(r.type){
				case START:
					remoteActivities.put(dac.getActivityId(), dac);
					break;
				case INPUT:
					sendInputData(dac, r.data);
					break;
				case CLOSE_INPUT:
					sendCloseInputData(dac, (Throwable)r.data);
					break;
				case OUTPUT:
					{
						ComputerId serverId = dac.getClientId();
						NetworkPacket packet = createWriteData(dac, r.data, r.type);
						networkManager.sendData(serverId, packet);
					}
					break;
				case CLOSE_OUTPUT:
					{
						ComputerId serverId = dac.getClientId();
						NetworkPacket packet = createCloseData(dac, (Throwable)r.data, r.type);
						networkManager.sendData(serverId, packet);
					}	
					break;
				}
			} catch (Throwable t) {
				failActivity(r.getDistributedActivityContext(), t);
			}
		}
	}
	private void processRemoteActivities(){
		//process input channels for remote activities
		DistributedActivityContext[] dacs = remoteActivities.values().toArray(new DistributedActivityContext[0]);
		for(DistributedActivityContext dac : dacs){
			try{
				processInputChannel(dac);
			} catch (Throwable t) {
				failActivity(dac, t);
			}
			final Channel<?> out = dac.getOutput();
			if(out.isClosed()) remoteActivities.remove(dac.getActivityId());
		}
	}
	
	private void sendInputData(DistributedActivityContext r, Object data) throws IOException{
		NetworkPacket packet = createWriteData(r, data, RequestType.INPUT);
		ComputerId serverId = r.selectRemoteNode(data);
		if(!r.hasServerId(serverId)){
			//first time
			packet.setActivity(serializeObject(r.getActivity()));
			packet.setImportPackage(r.getDistributedActivity().getImportPackage());
			packet.setRequireBundle(r.getDistributedActivity().getRequireBundle());
		}
		networkManager.sendData(serverId, packet);
		r.addServerId(serverId);
	}

	private void sendCloseInputData(DistributedActivityContext r, Throwable cause) throws IOException{
		NetworkPacket packet = createCloseData(r, cause, RequestType.CLOSE_INPUT);
		ComputerId[] servers = r.getServerIds();
		for(ComputerId server: servers){
			try{
				networkManager.sendData(server, packet);
			} 
			catch (Exception e) {		}
		}
	}
	
	private void processInputChannel(DistributedActivityContext r) throws IOException{
		final Channel<?> channel = r.getInput();
		final int available = channel.size();
		if(available > 0){
			final Lock<?> rl = channel.getReadLock(1);
			if(rl!=null){
				Object data = null;
				try{
					data = rl.get(0);
				}finally{
					rl.unlock();
				}
				r.startRequest();
				sendInputData(r, data);
			}
		}
		if(channel.isClosed() && channel.isEmpty() && r.isAlive()){
			r.setAlive(false);
			sendCloseInputData(r, channel.getFailureCause());
		}
	}
	
	private static void failActivity(DistributedActivityContext r, Throwable t){
		r.kill(t);
	}
	/**
	 * Fail all requests
	 * */
	public void failAllRequests(){
		DistributedActivityContext[] rs = null;
		final ReentrantLock lock = this.lock;
		lock.lock();
		try{
			rs = pendingRequests.toArray(new DistributedActivityContext[0]);
			for(DistributedActivityContext r : rs){
				failActivity(r, new IOException("Server shutdown in progress"));
			}
			pendingRequests.clear();
		} finally{
			lock.unlock();
		}
		rs = localActivities.toArray(new DistributedActivityContext[0]);
		for(DistributedActivityContext r : rs){
			failActivity(r, new IOException("Server shutdown in progress"));
			try{
				processOutputChannel(r);
			} catch (Throwable t) {
				failActivity(r, t);
			}
		}
		localActivities.clear();
		rs = remoteActivities.values().toArray(new DistributedActivityContext[0]);
		for(DistributedActivityContext r : rs){
			failActivity(r, new IOException("Server shutdown in progress"));
		}
		remoteActivities.clear();
	}
	/**
	 * Remove requests and activities sent to dead servers or serving from those
	 * */
	public void removeRequests(Computer c) {
		// remove requests sent to dead servers or serving from those
		final ComputerId id = c.getComputerId();
		DistributedActivityContext[] rs = null;

		final ReentrantLock lock = this.lock;
		lock.lock();
		try{
			rs = pendingRequests.toArray(new DistributedActivityContext[0]);
			for(DistributedActivityContext r : rs){
				if(r.hasServerId(id)){
					failActivity(r, new IOException("Server has left the cluster: "+id));
					pendingRequests.remove(r);
				}
			}
		} finally {
			lock.unlock();
		}
		rs = localActivities.toArray(new DistributedActivityContext[0]);
		for(DistributedActivityContext r : rs){
			if(r.getClientId().equals(id)){
				failActivity(r, new IOException("Client left the cluster: "+id));
				localActivities.remove(r);
			}
		}
		rs = remoteActivities.values().toArray(new DistributedActivityContext[0]);
		for(DistributedActivityContext r : rs){
			if(r.hasServerId(id)){
				failActivity(r, new IOException("Server has left the cluster: "+id));
				remoteActivities.remove(r);
			}
		}
			
	}
	private byte[] serializeObject(Object data){
		final Output out = new Output(1024, maxBufferSize);
		kryo.writeClassAndObject(out, data);
		return out.toBytes();
	}
	private NetworkPacket createWriteData(DistributedActivityContext dac, Object data, RequestType type){
		byte[] array = serializeObject(data);
		NetworkPacket rd=new NetworkPacket(dac.getActivityId(), type, array);
		return rd;
	}
	private NetworkPacket createCloseData(DistributedActivityContext dac, Throwable failed, RequestType type){
		byte[] data = serializeObject(failed);
		NetworkPacket rd=new NetworkPacket(dac.getActivityId(), type, data);
		return rd;
	}

	public NetworkManager getNetworkManager() {
		return networkManager;
	}

	public void setNetworkManager(NetworkManager networkManager) {
		this.networkManager = networkManager;
	}
	

}
